package cn.net.wanji.utils;

import java.util.Properties;
import org.apache.kafka.clients.producer.*;

public class KafkaProducerUtil {
    private static final String TOPIC_NAME = "FENGTAI_ETC_VEHICLE";
    private static final String BOOTSTRAP_SERVERS = "10.102.1.164:9092"; // 例如：localhost:9092
    public static void startkafka(String message ){
        // 设置Kafka生产者的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建Kafka的生产者实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            // 发送一条消息到Kafka
//            String message = "{\"recordId\":\"G0005110010001101002023071917000007\",\"recordType\":\"B\",\"ERID\":\"G000511001000110100\",\"vehiclePicInfos\":[{\"hourBatchNo\":\"2023071917\",\"driveDir\":1,\"devType\":\"4\",\"headPicTime\":\"2023-07-19T17:39:32\",\"sidePicTime\":\"2023-07-19T17:39:33\",\"tailPicTime\":\"\",\"headLaneNum\":\"1\",\"sideLaneNum\":\"1\",\"tailLaneNum\":\"\",\"licenseCode\":\"沪ADN4445\",\"licenseColor\":4,\"vehicleSpeed\":104,\"identifyType\":1,\"vehicleLength\":415,\"vehicleWidth\":-1,\"vehicleHeight\":-1,\"vehicleAxleCount\":2,\"vehicleAxleInfo\":\"11\",\"dangerousGoods\":2}],\"RSUInfos\":[{\"etcMediaType\":\"OBU\",\"etcMac\":\"55067989\",\"etcContractSerial\":\"4501207800424331\",\"etcTransTime\":\"2023-07-19T17:39:32\",\"etcTradeResult\":\"B500\",\"etcLicense\":\"沪ADN4445\",\"etcLicenseColor\":\"4\",\"etcVehClass\":\"1\",\"etcVehUserType\":\"0\",\"etcStatus\":\"2001\",\"etcValidSign\":\"AA\",\"etcEntryClass\":\"1\",\"etcEntryAxles\":\"2\",\"etcEntryTime\":\"2023-07-19T15:26:52\",\"etcGeneralState\":\"3\",\"etcTollStation\":\"3101090108\",\"vehTradeResult\":\"1\",\"relateEtcCount\":\"0\",\"relateEtcMediaType\":\"\",\"relateEtcMac\":\"\",\"relateEtcContractSerial\":\"\"}],\"unlawfulMold\":\"\",\"unlawfulCredible\":\"\"}";
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, message);
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("Error sending message to Kafka: " + exception.getMessage());
                    } else {
                        System.out.println("Message sent to Kafka, partition: " + metadata.partition() + ", offset: " + metadata.offset());
                    }
                }
            });

            // 等待消息发送完成
            producer.flush();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭Kafka生产者
            producer.close();
        }
    }
    public static void main(String[] args) {
        // 设置Kafka生产者的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建Kafka的生产者实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            // 发送一条消息到Kafka
            String message = "{\"recordId\":\"G0005110010001101002023071917000007\",\"recordType\":\"B\",\"ERID\":\"G000511001000110100\",\"vehiclePicInfos\":[{\"hourBatchNo\":\"2023071917\",\"driveDir\":1,\"devType\":\"4\",\"headPicTime\":\"2023-07-19T17:39:32\",\"sidePicTime\":\"2023-07-19T17:39:33\",\"tailPicTime\":\"\",\"headLaneNum\":\"1\",\"sideLaneNum\":\"1\",\"tailLaneNum\":\"\",\"licenseCode\":\"沪ADN4445\",\"licenseColor\":4,\"vehicleSpeed\":104,\"identifyType\":1,\"vehicleLength\":415,\"vehicleWidth\":-1,\"vehicleHeight\":-1,\"vehicleAxleCount\":2,\"vehicleAxleInfo\":\"11\",\"dangerousGoods\":2}],\"RSUInfos\":[{\"etcMediaType\":\"OBU\",\"etcMac\":\"55067989\",\"etcContractSerial\":\"4501207800424331\",\"etcTransTime\":\"2023-07-19T17:39:32\",\"etcTradeResult\":\"B500\",\"etcLicense\":\"沪ADN4445\",\"etcLicenseColor\":\"4\",\"etcVehClass\":\"1\",\"etcVehUserType\":\"0\",\"etcStatus\":\"2001\",\"etcValidSign\":\"AA\",\"etcEntryClass\":\"1\",\"etcEntryAxles\":\"2\",\"etcEntryTime\":\"2023-07-19T15:26:52\",\"etcGeneralState\":\"3\",\"etcTollStation\":\"3101090108\",\"vehTradeResult\":\"1\",\"relateEtcCount\":\"0\",\"relateEtcMediaType\":\"\",\"relateEtcMac\":\"\",\"relateEtcContractSerial\":\"\"}],\"unlawfulMold\":\"\",\"unlawfulCredible\":\"\"}";
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, message);
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("Error sending message to Kafka: " + exception.getMessage());
                    } else {
                        System.out.println("Message sent to Kafka, partition: " + metadata.partition() + ", offset: " + metadata.offset());
                    }
                }
            });

            // 等待消息发送完成
            producer.flush();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭Kafka生产者
            producer.close();
        }
    }
}

